package cn.azzhu.day02.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

/**
 * 并行为1的source
 *  env.socketTextStream("",9999)
 *  env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9)
 *  env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9))
 * @author azzhu
 * @create 2020-09-17 23:36:36
 */
public class SourceDemo1 {
    public static void main(String[] args) throws Exception {
        //实时计算环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建抽象的数据集【创建原始的抽象数据集的方法：Source】
        //DataStream是一个抽象的数据集
        //DataStream<String> socketTextStream = env.socketTextStream("",9999);

        //将客户端的集合并行化成一个抽象的数据集，通常用来做测试和实验
        //fromElements 是一个有界数据流，虽然是一个实时计算程序，但是数据处理完，程序就会退出
       // final DataStreamSource<Integer> nums = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9);
        //并行度为1的source
        final DataStreamSource<Integer> nums = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9));

        //获取这个DataStream的并行度
        final int parallelism = nums.getParallelism();
        System.out.println("==============>"+parallelism);  //todo 1

        final SingleOutputStreamOperator<Integer> filtered = nums.filter(num -> num % 2 == 0);

        final int parallelism1 = filtered.getParallelism();
        System.out.println("&&&&&&&&&&& >>>>" + parallelism1);  //todo 8 ,电脑的线程数
        filtered.print();

        env.execute("SourceDemo1");
    }
}
